package consume

import (
	"fmt"
	"github.com/bytedance/sonic"
	"huijing_mall/internal/mq"
	"huijing_mall/logic/collectlogic"
	"huijing_mall/pkg/logger"
	"huijing_mall/pkg/sendEmail"
	"huijing_mall/shared/get_ctx_value"
	"huijing_mall/shared/kernel/ask"
	"huijing_mall/variable"
	"os"
	"os/signal"
	"strings"
	"sync"
	"syscall"
	"time"
)

// ------------------------------
// 1. RabbitMQ Consume 方法（优化内存+退出逻辑）
// ------------------------------

// ------------------------------
// 2. 消费者启动逻辑（结构清晰，生命周期明确）
// ------------------------------

func StartMQConsumers(mq *mq.RabbitMQ, dbs *get_ctx_value.DBS) {
	const consumerCount = 3 // 每个队列的消费者数量
	var (
		wg        sync.WaitGroup    // 仅用于等待"正在处理的消息"完成
		stopFuncs []func(wait bool) // 存储所有消费者的停止函数
	)

	// 第二步：启动所有队列的消费者（模块化，结构清晰）
	// 2.1 启动"发送队列"消费者
	if err := startQueueConsumers(mq, variable.SendQu, consumerCount, &wg, &stopFuncs, func(body []byte) error {
		defer wg.Done() // 仅标记"当前消息处理完成"，不影响消费者监听
		logger.Info(fmt.Sprintf("发送队列消费者: 处理消息: %s", string(body)))

		// 业务逻辑：发送邮件/第三方请求
		if strings.Contains(string(body), "@") {
			bodyData := strings.Split(string(body), "_")
			if len(bodyData) >= 2 {
				_ = sendEmail.SendEmail([]string{bodyData[0]}, bodyData[1], "")
			}
		}
		time.Sleep(time.Second) // 模拟处理耗时
		return nil
	}); err != nil {
		logger.Error(fmt.Sprintf("启动发送队列消费者失败: %v", err))
		return
	}

	// 2.2 启动"任务队列"消费者
	if err := startQueueConsumers(mq, variable.TaskQu, consumerCount, &wg, &stopFuncs, func(body []byte) error {
		defer wg.Done()
		logger.Info(fmt.Sprintf("任务队列消费者: 处理消息: %s", string(body)))

		// 业务逻辑：解析消息+数据库操作
		bodyData := strings.Split(string(body), "_")
		if len(bodyData) < 3 {
			return fmt.Errorf("消息格式错误: %s", string(body))
		}
		var cb ask.CityDBRequest
		if err := sonic.Unmarshal([]byte(bodyData[2]), &cb); err != nil {
			return fmt.Errorf("解析消息失败: %v", err)
		}
		err, _, ok := collectlogic.FindOne(dbs, cb)
		if err != nil && !ok {
			return err
		}
		time.Sleep(time.Second)
		return nil
	}); err != nil {
		logger.Error(fmt.Sprintf("启动任务队列消费者失败: %v", err))
		return
	}
	var (
		sigChan = make(chan os.Signal, 1) // 信号通道（缓冲避免阻塞）
	)

	// 第一步：注册信号监听（仅监听Ctrl+C和kill信号）
	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
	defer signal.Stop(sigChan) // 程序退出前取消信号监听

	// 第三步：阻塞等待Ctrl+C信号（核心！确保消费者持续监听）
	logger.Info("所有消费者启动完成，服务运行中（Ctrl+C退出）")
	<-sigChan // 这里会一直阻塞，直到收到Ctrl+C
	logger.Info("收到退出信号，开始优雅关闭")

	// 第四步：优雅关闭流程
	// 4.1 停止所有消费者（不再接收新消息）
	for _, stopFunc := range stopFuncs {
		stopFunc(true) // 等待消费者完全退出
	}

	// 4.2 等待所有"正在处理的消息"完成（超时30秒）
	logger.Info("等待正在处理的消息完成...")
	doneChan := make(chan struct{}, 1)
	go func() {
		wg.Wait()
		doneChan <- struct{}{}
	}()
	select {
	case <-doneChan:
		logger.Info("所有消息处理完成")
	case <-time.After(30 * time.Second):
		logger.Warn("等待消息处理超时，强制关闭（未完成消息会重新入队）")
	}

	// 4.3 关闭MQ连接
	if err := mq.Close(); err != nil {
		logger.Error("关闭MQ连接失败: %v", err)
	} else {
		logger.Info("MQ连接已关闭")
	}

	logger.Info("服务已优雅退出")
}

func startQueueConsumers(
	mq *mq.RabbitMQ,
	queueName string,
	count int,
	wg *sync.WaitGroup,
	stopFuncs *[]func(wait bool),
	handler func([]byte) error,
) error {
	// 先声明队列（确保队列存在，避免消费失败）
	if err := mq.DeclareQueue(queueName); err != nil {
		return fmt.Errorf("声明队列[%s]失败: %v", queueName, err)
	}

	// 启动count个消费者
	for i := 0; i < count; i++ {
		consumerID := fmt.Sprintf("%s_consumer_%d", queueName, i)
		// 每次处理消息前，先增加WaitGroup计数（确保不遗漏正在处理的消息）
		wg.Add(1)
		stopFunc, err := mq.Consume(queueName, consumerID, handler)
		if err != nil {
			wg.Done() // 启动失败，回滚计数
			logger.Error("启动消费者%s失败: %v", consumerID, err)
			continue
		}
		*stopFuncs = append(*stopFuncs, stopFunc)
	}

	logger.Info(fmt.Sprintf("队列[%s]已启动%d个消费者", queueName, count))
	return nil
}
